package net.xuele.learn.IO.nio;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Date;
import java.util.Iterator;
import java.util.Set;

/**
 * Created by GaoQingming on 2019/4/11 0011.
 * <p>
 * <p>
 * 开发的流程：
 * 1、打开ServerSocketChannel，监听客户端的连接，所有客户端连接的父管道
 * 2、绑定监听的端口，设置连接为非阻塞模式
 * 3、创建Reactor线程，创建多路复用器并启动线程
 * 4、将ServerSocketChannel注册到Reactor线程的多路复用器上，监听accept（连接）事件
 * 5、多路复用器在线程的run方法的无限循环体内轮询准备就绪的Key
 * 6、当多路复用器监听到有新的客户端接入，处理新的接入请求，完成TCP三次握手，建立物理链路
 * 7、设置客户端为非阻塞模式
 * 8、将新接入的客户端连接注册到Reactor线程的多路复用器上，监听读操作，读取客户端发送过来的网络消息
 * 9、异步读取客户端的消息到缓冲区
 * 10、对ByteBuffer进行编解码，如果有半包消息指针reset，继续读取后续的报文，将解码成功的消息封装成Task，
 * 投递到业务线程池，进行业务逻辑处理、
 * 11、将对象编码成ByteBuffer，调用SocketChannel的异步write接口，将消息异步发送给客户端。
 */
public class MultipleTimeServer implements Runnable {


    private Selector selector;

    private ServerSocketChannel serverSocketChannel;

    private volatile boolean stop;

    public MultipleTimeServer(int port) {
        try {
            // 创建多路复用器
            selector = Selector.open();
            // 创建服务端的Channel
            serverSocketChannel = ServerSocketChannel.open();
            serverSocketChannel.configureBlocking(false);
            serverSocketChannel.socket().bind(new InetSocketAddress(port), 1024);
            // 将ServerSocketChannel注册到多路复用器Selector上，并监听客户端连接事件
            serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
            System.out.println("the server started in port:" + port);
        } catch (Exception e) {
            e.printStackTrace();
            System.exit(1);
        }

    }


    public void stop() {
        this.stop = true;
    }

    @Override
    public void run() {
        // 在无限循环中遍历selector

        /**
         *  关于jdk的epoll bug：即selector空轮询，导致CPU100%的问题！https://www.jianshu.com/p/3ec120ca46b2
         *
         *
         */

        while (!stop) {
            try {
                // 每隔一秒进行一次轮询，无论是否有读写事件，selector休眠1s后都会被唤醒
                // 线程没有睡眠
                /**
                 * 正常情况下，selector.select的操作是阻塞的，只有注册在上面的channel发生了读写操作才会被唤醒
                 * bug的产生就是没有任何读写事件，但是selector被唤醒了，很显然SelectorKey是一个空的集合，
                 * 于是遍历SelectorKey集合的操作不会执行，相反，就会执行上面的while循环，最终导致CPU100%
                 *
                 * bug产生的原因：具体解释为：在部分Linux的2.6的kernel中，
                 * poll和epoll对于突然中断的连接socket会对返回的eventSet事件集合置为POLLHUP，也可能是POLLERR，
                 * eventSet事件集合发生了变化，这就可能导致Selector会被唤醒。
                 *
                 * 终极的解决办法是：重新创建一个Selector，并将旧的销毁
                 * Jetty解决方案如下：
                 * 定义了两个参数，一个是select返回值为0 的计数，第二个是多长时间，整体就是控制在多长时间内，如果selector.select
                 * 不断返回0，那么就进入jvm的bug模式
                 *
                 * 做法是：
                 *  1、记录select()返回0 的次数（记做jvmBug次数）
                 *  2、在MONITOR_PERIOD时间范围内，如果jvmBug次数超过JVMBUG_THRESHHOLD，则新建一个Selector，
                 *  并且需要复制SelectionKey中的关注事件到新的Selector中
                 *
                 *  Netty的做法：
                 *              long currentTimeNanos = System.nanoTime();
                 for (;;) {
                 // 1.定时任务截止事时间快到了，中断本次轮询
                 ...
                 // 2.轮询过程中发现有任务加入，中断本次轮询
                 ...
                 // 3.阻塞式select操作
                 selector.select(timeoutMillis);
                 // 4.解决jdk的nio bug
                 long time = System.nanoTime();
                 if (time - TimeUnit.MILLISECONDS.toNanos(timeoutMillis) >= currentTimeNanos) {
                 selectCnt = 1;
                 } else if (SELECTOR_AUTO_REBUILD_THRESHOLD > 0 &&
                 selectCnt >= SELECTOR_AUTO_REBUILD_THRESHOLD) {

                 rebuildSelector();
                 selector = this.selector;
                 selector.selectNow();
                 selectCnt = 1;
                 break;
                 }
                 currentTimeNanos = time;
                 ...
                 }
                 *  netty 会在每次进行 selector.select(timeoutMillis) 之前记录一下开始时间currentTimeNanos，
                 *  在select之后记录一下结束时间，判断select操作是否至少持续了timeoutMillis秒
                 *  （这里将time - TimeUnit.MILLISECONDS.toNanos(timeoutMillis) >= currentTimeNanos改成time - currentTimeNanos >= TimeUnit.MILLISECONDS.toNanos(timeoutMillis)或许更好理解一些）,
                 *  如果持续的时间大于等于timeoutMillis，说明就是一次有效的轮询，重置selectCnt标志，
                 *  否则，表明该阻塞方法并没有阻塞这么长时间，可能触发了jdk的空轮询bug，
                 *  当空轮询的次数超过一个阀值的时候，默认是512，就开始重建selector
                 *
                 *  ————给select方法一个时间，如果某次select操作的事件小于这个值，则有可能出现了jdk的bug
                 */
                selector.select(1000);
                // 轮询出来的结果，但是不一定有，SelectionKey中有已经就绪的Channel（有读写事件）！！！
                Set<SelectionKey> selectionKeys = selector.selectedKeys();
                Iterator<SelectionKey> iterator = selectionKeys.iterator();
                SelectionKey key;
                // 遍历就绪的通道集合
                while (iterator.hasNext()) {
                    key = iterator.next();
                    iterator.remove();
                    try {
                        // 处理就绪的Channel
                        handleInput(key);
                    } catch (Exception e) {
                        if (key.channel() != null) {
                            key.channel().close();
                        }
                    }
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        // 多路复用器关闭后，所有注册在上面的Channel都会被自动去注册和关闭，所以不需要去重复释放资源
        if (selector != null) {
            try {
                selector.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    // 处理SelectionKey中准备继续的Channel，很显然，连接请求，准备就绪的Channel是ServerSocketChannel（有连接请求过来）
    // 而读请求则由客户端过来的，所以使用的Channel是SocketChannel
    private void handleInput(SelectionKey key) throws Exception {
        if (key.isValid()) {
            // 处理接入的请求！！！
            if (key.isAcceptable()) {
                // 接入的请求的事件发生在ServerSocketChannel上
                ServerSocketChannel ssc = (ServerSocketChannel) key.channel();
                SocketChannel sc = ssc.accept();
                System.out.println("====accept======="+sc.hashCode());
                sc.configureBlocking(false);
                // 连接成功，需要监听SocketChannel上的读请求
                sc.register(selector, SelectionKey.OP_READ);
            }

            // 读取数据
            if (key.isReadable()) {
                SocketChannel sc = (SocketChannel) key.channel();
                System.out.println("======read====="+sc.hashCode());
                ByteBuffer buffer = ByteBuffer.allocate(1024);
                int read = sc.read(buffer);
                if (read > 0) {
                    // 将缓冲区当前的limit设置为position，position设置为0，用于后续对缓冲区的读取操作
                    buffer.flip();
                    // 创建一个与buffer中数据长度相同的字节数组
                    byte[] bytes = new byte[buffer.remaining()];
                    // 将缓冲区数据读入到数组中
                    buffer.get(bytes);
                    String body = new String(bytes, "UTF-8");
                    System.out.println("server received msg:" + body);
                    String currentTime = "QUERY TIME ORDER".equalsIgnoreCase(body) ? new Date(System.currentTimeMillis()).toString() : "ERROR";
                    doWrite(sc, currentTime);
                } else if (read < 0) {
                    key.cancel();
                    sc.close();
                } else {
                    //读到0字节，忽略
                }
            }
        }
    }

    private void doWrite(SocketChannel sc, String msg) throws IOException {
        if (msg != null && msg.length() > 0) {
            // 发送数据到客户端，需要先将数据编码成字节数组
            byte[] bytes = msg.getBytes();
            ByteBuffer allocate = ByteBuffer.allocate(bytes.length);
            // 将字节数组数据复制到缓冲区
            allocate.put(bytes);

            allocate.flip();
            // SocketChannel是异步非阻塞的，
            sc.write(allocate);
        }
    }
}
